package day03.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author wsl
  * @version
  * 1.只增加checkpoint，没有增加Cache缓存打印
  * 第1个job执行完，触发了checkpoint，另起一个job从头计算一遍，并把数据存储在检查点上，做验算，切断血缘关系。
  * 第3、4个job，数据从检查点上直接读取。
  *
  * 2.增加checkpoint，也增加Cache缓存打印
  * 第1个job执行完，数据就保存到Cache里面了，触发了checkpoint，另起一个job直接读取Cache里面的数据，并把数据存储在检查点上。
  * 第3、4个job，数据从检查点上直接读取。
 *
 * cache与ck区别：
 *
  */
object CheckPoint {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //设置检查点路径
    sc.setCheckpointDir("./checkpoint")
   // sc.setCheckpointDir("hdfs://hadoop102:8020/checkpoint")

    val wordToOneRdd: RDD[(String, Long)] = sc.textFile("sparkcore/input/1.txt")
      .flatMap(line => line.split(" "))
      .map {
        word => {
          (word, System.currentTimeMillis())
        }
      }


    //wordToOneRdd.cache()

    wordToOneRdd.checkpoint()

    //释放缓存 sc.stop也会释放
    wordToOneRdd.unpersist()

    //触发执行
    wordToOneRdd.collect().foreach(println)

    //再次触发
    wordToOneRdd.collect().foreach(println)
    wordToOneRdd.collect().foreach(println)

    Thread.sleep(10000000)
    sc.stop()

  }
}
